package com.dahantc;

import lombok.extern.slf4j.Slf4j;

import java.util.HashMap;
import java.util.Map;
import java.util.Objects;

/**
 * kafka测试程序：测试指定kafka的收发联通性
 * @author ydq
 * @date 2022/9/29
 */
@Slf4j
public class KafkaApp {

    public static String[] TOPIC_CONSUMERS;
    public static String[] TOPIC_PRODUCERS;
    public static String KAFKA_URL;
    public static String KAFKA_USER;
    public static String KAFKA_PASSWORD;
    public static String IS_NEED_PWD = "F";
    public static String MECHANISM = "PLAIN";

    private static final String[] ARGS_NAMES = {"tp_consumers", "tp_producers", "kf_url"};
    private static final String[] SELECT_ARGS_NAMES = {"isneedpwd", "kf_user", "kf_pwd", "mechanism"};


    public static void main( String[] args) {
        log.info("==============================================================");
        log.info("=****   kafka demo for test consumer&producer            ****=");
        log.info("==============================================================");
        if (null == args || args.length == 0) {
           throw new IllegalArgumentException("Illegal argument, next Args are necessary: " + String.join(",", ARGS_NAMES));
        }
        log.info("Kafka with [protocol=SASL_PLAINTEXT] connect test program is starting...");
        handleArgs(args);
        for (String _cTopic : TOPIC_CONSUMERS) {
            new TaskConsumer(_cTopic, 3).start();
        }
        for (String _pTopic : TOPIC_PRODUCERS) {
            new TaskProducer(_pTopic, 3).start();
        }
    }

    private static void handleArgs(String[] args) {
        Map<String, String> _argMap = new HashMap<>();
        for (String arg : args) {
            String[] _arr = arg.split("=");
            if (_arr.length != 2) {
                log.error("Error arg: " + arg);
                continue;
            }
            _argMap.put(_arr[0].toLowerCase(), _arr[1]);
        }
        // handle arg map 2 static
        for (String argName : ARGS_NAMES) {
            String argVal = _argMap.get(argName);
            if (Objects.isNull(_argMap.get(argName)) || argVal.trim().length() == 0) {
                throw new IllegalArgumentException("Needed arg " + argName + " is missing!");
            }
        }
        setArgs(_argMap);
        _argMap.clear();
        _argMap = null;
    }

    private static void setArgs(Map<String, String> argMap) {
        log.info("SetArgs: {}", argMap);
        TOPIC_CONSUMERS = argMap.get(ARGS_NAMES[0]).split(",");
        TOPIC_PRODUCERS = argMap.get(ARGS_NAMES[1]).split(",");
        KAFKA_URL = argMap.get(ARGS_NAMES[2]);
        IS_NEED_PWD = argMap.get(SELECT_ARGS_NAMES[0]);
        if ("T".equals(IS_NEED_PWD)) {
            KAFKA_USER = argMap.get(SELECT_ARGS_NAMES[1]);
            KAFKA_PASSWORD = argMap.get(SELECT_ARGS_NAMES[2]);
            MECHANISM = Objects.nonNull(argMap.get(SELECT_ARGS_NAMES[3])) ? argMap.get(SELECT_ARGS_NAMES[3]) : MECHANISM;
        }
    }

}
